package tsif.tcluster.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import tsif.tcluster.TClusterContext;
import tsif.tcluster.netty.codec.TClusterNettyMessage;
import tsif.tcluster.rpc.TClusterRpcProtocol;
import tsif.tcluster.rpc.TClusterRpcUtils;
import tsif.tcluster.session.TClusterSession;

/**
 * 用netty实现的集群session基类<br>
 * 
 */
public class TClusterNettySession extends TClusterSession {
	protected Channel ch;

	public TClusterNettySession(TClusterContext context, int targetContextId, Channel ch) {
		super(context, targetContextId);
		this.ch = ch;
	}

	@Override
	public void send(long sourceServiceId, long targetServiceId, short code, Object data) {
		if (ch == null || !ch.isActive()) {
			throw new RuntimeException("no connect ! " + this + " targetContextId=" + this.targetContextId);
		}

		// date to byte
		byte[] dateByte = null;
		if (data.getClass() != byte[].class) {
			dateByte = TClusterRpcUtils.toByte(data, data.getClass());
		} else {
			dateByte = (byte[]) data;
		}

		// create packet
		TClusterNettyMessage packet = new TClusterNettyMessage();
		packet.setCode(code);
		packet.setData(dateByte);
		packet.setSourceServiceId(sourceServiceId);
		packet.setTargetServiceId(targetServiceId);

		// send packet
		ch.writeAndFlush(packet);
	}

	/** 接收消息 **/
	protected void recv(final TClusterNettyMessage packet) throws Exception {
		short code = packet.getCode();
		if (code == TClusterRpcProtocol.RPC_KEEP //
				|| code == TClusterRpcProtocol.RPC_INFO //
		//
		) {
			return; // 心跳处理
		}

		// 直接接收消息处理(这里提交线程池没意义)
		recv(packet.getSourceServiceId(), packet.getTargetServiceId(), packet.getCode(), packet.getData());
	}

	@Override
	public void close() {
		if (ch != null) {
			ch.close();
			ch = null;
		}
	}

	@Override
	public boolean alive() {
		return (ch != null) && ch.isActive();
	}

	/** session处理接口 **/
	public static class TClusterNettySessionHandler extends SimpleChannelInboundHandler<TClusterNettyMessage> {
		protected TClusterNettySession session;

		@Override
		protected void channelRead0(ChannelHandlerContext context, TClusterNettyMessage packet) throws Exception {
			session.recv(packet);
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
			if (cause.getClass() == java.io.IOException.class) {
				// io error
			} else {
				System.err.println("channel error! " + ctx.channel() + " " + cause);
				cause.printStackTrace();
			}
			ctx.channel().close();
		}

		public TClusterNettyContext getContext() {
			return (session != null) ? (TClusterNettyContext) session.getContext() : null;
		}

		public TClusterSession getSession() {
			return session;
		}

		public void setSession(TClusterNettySession session) {
			this.session = session;
		}

	}

}
